//! db_impl contains the implementation of the database interface and high-level compaction and
//! maintenance logic.

#![allow(unused_attributes)]

use crate::cache::Cache;
use crate::db_iter::DBIterator;

use crate::cmp::{Cmp, InternalKeyCmp};
use crate::env::{Env, FileLock};
use crate::error::{err, Result, StatusCode};
use crate::filter::{BoxedFilterPolicy, InternalFilterPolicy};
use crate::infolog::Logger;
use crate::key_types::{parse_internal_key, InternalKey, LookupKey, ValueType};
use crate::log::{LogReader, LogWriter};
use crate::memtable::MemTable;
use crate::merging_iter::MergingIter;
use crate::options::Options;
use crate::snapshot::{Snapshot, SnapshotList};
use crate::table_builder::TableBuilder;
use crate::table_cache::{table_file_name, TableCache};
use crate::types::{
    parse_file_name, share, FileMetaData, FileNum, FileType, LdbIterator, SequenceNumber, Shared,
    MAX_SEQUENCE_NUMBER, NUM_LEVELS,
};
use crate::version::Version;
use crate::version_edit::VersionEdit;
use crate::version_set::{
    manifest_file_name, read_current_file, set_current_file, Compaction, VersionSet,
};
use crate::write_batch::WriteBatch;

use bytes::Bytes;
use std::cmp::Ordering;
use std::io::{self, BufWriter, Write};
use std::mem;
use std::ops::Drop;
use std::path::Path;
use std::path::PathBuf;
use std::rc::Rc;

/// DB contains the actual database implemenation. As opposed to the original, this implementation
/// is not concurrent (yet).
pub struct DB {
    /// 数据库的名称（可能用于标识或日志输出）。
    name: PathBuf,
    /// 数据库在文件系统中的存储路径（用于定位数据文件、日志文件等）。
    path: PathBuf,
    /// 数据库的文件锁，用于确保同一时间只有一个进程访问该数据库（防止并发冲突）。Option 表示可能未获取锁（如初始化阶段）。
    lock: Option<FileLock>,

    /// 内部键比较器，实现 Cmp 特性（自定义比较逻辑），用于对数据库中的键（包括版本号等元数据）进行排序和比较。Rc 包装表示可能被多个组件共享。
    internal_cmp: Rc<Box<dyn Cmp>>,
    /// 内部过滤策略（如布隆过滤器），用于优化读取性能，快速判断一个键是否存在于 SSTable 中。
    fpol: InternalFilterPolicy<BoxedFilterPolicy>,
    /// 数据库的全局配置选项（如块大小、压缩方式、写入缓冲区大小等）。
    opt: Options,

    /// 当前活跃的内存表，新写入的数据首先会存储在这里（写入速度快，内存中的有序数据结构）。
    mem: MemTable,
    ///  immutable（不可变）内存表。当 mem 达到阈值时，会被转换为 imm 并等待后台线程将其持久化为 SSTable。Option 表示可能不存在（如刚启动时）。
    imm: Option<MemTable>,

    /// 日志写入器，用于将写入操作记录到 WAL（Write-Ahead Log）日志文件，确保崩溃后可恢复内存表中的数据。Option 表示可能未打开日志（如初始化阶段）。
    log: Option<LogWriter<BufWriter<Box<dyn Write>>>>,
    /// 当前活跃日志文件的编号（与 log 对应，用于标识日志文件）。
    log_num: Option<FileNum>,
    /// SSTable 文件的缓存，用于缓存打开的 SSTable 句柄和数据块，减少磁盘 I/O 提升读取性能。Shared 表示可能被多个线程共享。
    cache: Shared<TableCache>,
    /// 数据库版本集合，管理所有 SSTable 的版本信息、文件编号和压缩状态（之前分析过的核心元数据组件）。
    vset: Shared<VersionSet>,
    /// 快照列表，存储数据库的历史快照（用于实现多版本并发读取，MVCC），每个快照对应一个特定的序列号。
    snaps: SnapshotList,

    /// 每个层级（Level）的压缩统计信息数组，记录压缩操作的次数、数据量等指标（用于监控和优化压缩策略）。
    cstats: [CompactionStats; NUM_LEVELS],
}

unsafe impl Send for DB {}

impl DB {
    // RECOVERY AND INITIALIZATION //

    /// new initializes a new DB object, but doesn't touch disk.
    fn new<P: AsRef<Path>>(name: P, mut opt: Options) -> DB {
        let name = name.as_ref();
        if opt.log.is_none() {
            let log = open_info_log(opt.env.as_ref().as_ref(), name);
            opt.log = Some(share(log));
        }
        let path = name.canonicalize().unwrap_or(name.to_owned());

        let cache = share(TableCache::new(
            name,
            opt.clone(),
            share(Cache::new(opt.block_cache_capacity_bytes / opt.block_size)),
            opt.max_open_files - 10,
        ));
        let vset = VersionSet::new(name, opt.clone(), cache.clone());

        DB {
            name: name.to_owned(),
            path,
            lock: None,
            internal_cmp: Rc::new(Box::new(InternalKeyCmp(opt.cmp.clone()))),
            fpol: InternalFilterPolicy::new(opt.filter_policy.clone()),

            mem: MemTable::new(opt.cmp.clone()),
            imm: None,

            opt,

            log: None,
            log_num: None,
            cache,
            vset: share(vset),
            snaps: SnapshotList::new(),

            cstats: Default::default(),
        }
    }

    /// 这段代码定义了一个 current 方法，用于获取当前的 Version 共享引用（Shared<Version>）。
    ///
    /// 方法通过 self.vset.borrow() 获取对 vset 字段的访问（vset 很可能是一个 RefCell 包裹的版本集合 / 管理器）
    ///
    /// 然后调用该版本集合的 current() 方法（即前文中提到的 current 方法），获取当前版本的共享引用并返回
    fn current(&self) -> Shared<Version> {
        self.vset.borrow().current()
    }

    /// Opens or creates a new or existing database. `name` is the name of the directory containing
    /// the database.
    ///
    /// Whether a new database is created and what happens if a database exists at the given path
    /// depends on the options set (`create_if_missing`, `error_if_exists`).
    pub fn open<P: AsRef<Path>>(name: P, opt: Options) -> Result<DB> {
        let name = name.as_ref();
        let mut db = DB::new(name, opt);
        let mut ve = VersionEdit::new();
        let save_manifest = db.recover(&mut ve)?;

        // Create log file if an old one is not being reused.
        // 如果旧的那个文件不被使用就重新打开一个新的
        if db.log.is_none() {
            let lognum = db.vset.borrow_mut().new_file_number();
            let logfile = db.opt.env.open_writable_file(Path::new(&log_file_name(&db.name, lognum)))?;
            ve.set_log_num(lognum);
            db.log = Some(LogWriter::new(BufWriter::new(logfile)));
            db.log_num = Some(lognum);
        }

        // 是需要保存元数据变更
        if save_manifest {
            // 设置日志文件的编号
            ve.set_log_num(db.log_num.unwrap_or(0));
            // 设置新版本与应用新版本 和 持久化元数据
            db.vset.borrow_mut().log_and_apply(ve)?;
        }
        // 删除在文件系统中长时间不用的文件
        db.delete_obsolete_files()?;
        db.maybe_do_compaction()?;
        Ok(db)
    }

    /// initialize_db initializes a new database.
    ///
    /// 这个 initialize_db 方法是全新数据库的初始化逻辑，用于在首次创建数据库时，设置基础目录结构、初始化元数据文件（Manifest）和版本信息，
    /// 为数据库的正常运行奠定基础。它是数据库 “从无到有” 的创建入口。
    fn initialize_db(&mut self) -> Result<()> {
        // VersionEdit 用于记录数据库版本的变更信息，这里初始化了新数据库的基础元数据：
        // 比较器：决定键的排序规则（影响 SSTable 的组织和查询）；
        // 文件编号：数据库中所有文件（日志、SSTable、Manifest 等）都有唯一编号，从 2 开始分配（1 预留给首个 Manifest 文件）；
        // 序列号：跟踪操作顺序，初始为 0（尚无任何写入操作）。
        let mut ve = VersionEdit::new();
        ve.set_comparator_name(self.opt.cmp.id()); // 设置比较器名称（用于键排序）
        ve.set_log_num(0);  // 初始日志文件编号为 0
        ve.set_next_file(2); // 下一个可用的文件编号为 2（1 已用于 Manifest）
        ve.set_last_seq(0);  // 初始序列号为 0（无任何操作）
        // Manifest 文件是数据库的 “版本日志”，记录所有版本变更历史（如 SSTable 的创建、删除、层级变化等），是恢复数据库状态的关键；
        // 新数据库的首个 Manifest 文件编号为 1，包含初始版本的元数据（从 VersionEdit 写入）。
        {
            let manifest = manifest_file_name(&self.path, 1); // 生成 Manifest 文件名（如 "MANIFEST-000001"）
            let manifest_file = self.opt.env.open_writable_file(Path::new(&manifest))?; // 打开可写文件
            let mut lw = LogWriter::new(manifest_file); // 创建日志写入器
            lw.add_record(&ve.encode())?; // 将版本编辑信息编码后写入 Manifest
            lw.flush()?; // 强制刷盘，确保元数据持久化
        }
        // current 文件是一个 “指针文件”，记录当前有效的 Manifest 文件编号（这里是 1）；
        // 数据库启动时会先读取 current 文件，找到最新的 Manifest，再从中恢复版本信息；
        // 这一步确保后续操作能正确定位到当前的元数据。
        set_current_file(self.opt.env.as_ref().as_ref(), &self.path, 1)
    }

    /// recover recovers from the existing state on disk. If the wrapped result is `true`, then
    /// log_and_apply() should be called after recovery has finished.
    ///
    /// 这个 recover 方法是 LSM 树存储引擎中数据库启动时的恢复（Recovery）核心逻辑，负责从磁盘上的现有数据（包括日志文件、SSTable、版本元数据等）恢复数据库状态，
    /// 确保进程重启后数据不丢失，并验证数据完整性。它是数据库从 “离线状态” 到 “可用状态” 的关键过渡步骤。
    ///
    /// recover 方法的核心价值是实现数据库的崩溃恢复，确保：
    ///
    /// 数据不丢失：从预写日志（WAL）中恢复未被持久化到 SSTable 的操作（内存表数据）；
    ///
    /// 状态一致性：恢复版本集合（VersionSet）的元数据，确保各层级 SSTable 信息正确；
    ///
    /// 完整性校验：检查是否有缺失的关键文件，避免使用损坏的数据库；
    ///
    /// 无缝重启：将数据库恢复到崩溃前的状态，使其能够正常处理新的读写请求。
    fn recover(&mut self, ve: &mut VersionEdit) -> Result<bool> {
        // 检查数据库是否已存在（根据配置决定是否允许创建新数据库）
        if self.opt.error_if_exists && self.opt.env.exists(self.path.as_ref()).unwrap_or(false) {
            return err(StatusCode::AlreadyExists, "database already exists");
        }
        // 创建数据库目录（若不存在）
        let _ = self.opt.env.mkdir(Path::new(&self.path));
        // 获取数据库锁（防止多进程同时访问）
        self.acquire_lock()?;
        // current 文件是数据库元数据的入口，记录当前有效的版本信息（如最新的 Manifest 文件路径）；
        // 若 current 文件不存在且允许创建新数据库，调用 initialize_db 初始化全新的数据库结构。
        if let Err(e) = read_current_file(self.opt.env.as_ref().as_ref(), &self.path) {
            if e.code == StatusCode::NotFound && self.opt.create_if_missing {
                // 若元数据文件不存在且允许创建，初始化新数据库
                self.initialize_db()?;
            } else {
                return err(StatusCode::InvalidArgument, "database does not exist and create_if_missing is false", );
            }
        }

        // If save_manifest is true, we should log_and_apply() later in order to write the new
        // manifest.
        let mut save_manifest = self.vset.borrow_mut().recover()?;

        // Recover from all log files not in the descriptor.
        let mut max_seq = 0;
        // 获取指定文件夹（p）下的所有直接子项（文件或子文件夹）的名称 disk_env
        let filenames = self.opt.env.children(&self.path)?;
        // 获取当前版本中记录的“活跃文件”（已被处理的 SSTable 等）
        let mut expected = self.vset.borrow().live_files();
        let mut log_files = vec![];

        for file in &filenames {
            // 获取 (文件编号, 文件类型)
            if let Ok((num, typ)) = parse_file_name(file) {
                expected.remove(&num);  // 移除已知的活跃文件
                // 筛选出需要恢复的日志文件（未被处理的 WAL）
                if typ == FileType::Log && (num >= self.vset.borrow().log_num || num == self.vset.borrow().prev_log_num) {
                    log_files.push(num);
                }
            }
        }
        // 检查是否有缺失的活跃文件（可能是数据损坏）
        if !expected.is_empty() {
            log!(self.opt.log, "Missing at least these files: {:?}", expected);
            return err(StatusCode::Corruption, "missing live files (see log)");
        }
        // 按文件编号排序（保证按写入顺序恢复）
        log_files.sort();
        for i in 0..log_files.len() {
            // 恢复单个日志文件，返回是否需要保存元数据和日志中的最大序列号
            let (save_manifest_, max_seq_) = self.recover_log_file(log_files[i], i == log_files.len() - 1, ve)?;
            // 是否需要保存元数据变更
            if save_manifest_ {
                save_manifest = true;
            }
            // 日志中记录的最大序列号
            if max_seq_ > max_seq {
                max_seq = max_seq_;
            }
            self.vset.borrow_mut().mark_file_number_used(log_files[i]); // 标记文件编号已使用
        }
        // 确保全局序列号不小于日志中记录的最大序列号（保证版本正确性）
        if self.vset.borrow().last_seq < max_seq {
            self.vset.borrow_mut().last_seq = max_seq;
        }
        // 返回 是否需要保存元数据变更
        Ok(save_manifest)
    }

    /// recover_log_file reads a single log file into a memtable, writing new L0 tables if
    /// necessary. If is_last is true, it checks whether the log file can be reused, and sets up
    /// the database's logging handles appropriately if that's the case.
    ///
    /// recover_log_file 的作用是 “重放日志文件中的操作，恢复未持久化的数据”，具体包括：
    /// 读取日志文件中的批量操作记录。
    /// 将操作重放到内存表，恢复数据状态。
    /// 当内存表达到阈值时，将其转换为磁盘上的 SSTable。
    /// 根据条件决定是否复用日志文件，或清理并持久化剩余数据。
    fn recover_log_file(
        &mut self,
        log_num: FileNum,
        is_last: bool,
        ve: &mut VersionEdit,
    ) -> Result<(bool, SequenceNumber)> {
        // 把文件名拼上 .log 后缀
        let filename = log_file_name(&self.path, log_num);
        // 打开文件
        let logfile = self.opt.env.open_sequential_file(Path::new(&filename))?;
        // Use the user-supplied comparator; it will be wrapped inside a MemtableKeyCmp.
        let cmp: Rc<Box<dyn Cmp>> = self.opt.cmp.clone();
        // 创建日志读取器（LogReader），启用校验和验证
        let mut logreader = LogReader::new(
            logfile, // checksum=
            true,
        );
        log!(self.opt.log, "Recovering log file {:?}", filename);
        // 初始化临时缓冲区、内存表（MemTable）和批量写入容器（WriteBatch）
        let mut scratch = vec![];
        let mut mem = MemTable::new(cmp.clone());
        let mut batch = WriteBatch::new();

        let mut compactions = 0;
        let mut max_seq = 0;
        let mut save_manifest = false;
        // 获取记录完数据的长度, 逐条读取日志记录，解析为批量操作（WriteBatch），并将这些操作重放到内存表中，模拟崩溃前的写入过程。
        while let Ok(len) = logreader.read(&mut scratch) {
            if len == 0 {
                break;
            }
            // 日志文件长度小于12日志报错并跳过
            if len < 12 {
                log!(self.opt.log,"corruption in log file {:06}: record shorter than 12B",log_num);
                continue;
            }
            // 将读取的日志内容解析为WriteBatch
            batch.set_contents(&scratch);
            // 将batch中的操作插入到内存表（重放写入操作）
            batch.insert_into_memtable(batch.sequence(), &mut mem);
            // 跟踪最大序列号（保证操作顺序）
            let last_seq = batch.sequence() + batch.count() as u64 - 1;
            if last_seq > max_seq {
                max_seq = last_seq
            }
            // 当内存表达到阈值 write_buffer_size 时，将其转换为L0层的SSTable（持久化到磁盘）
            if mem.approx_mem_usage() > self.opt.write_buffer_size {
                compactions += 1;
                self.write_l0_table(&mem, ve, None)?;// 生成SSTable
                save_manifest = true;// 标记需要更新元数据
                mem = MemTable::new(cmp.clone());// 重置内存表
            }
            batch.clear();
        }

        // Check if we can reuse the last log file.
        // 处理日志文件的复用或清理
        // 条件：启用日志复用、是最后一个日志文件、且未发生压缩（内存表未被持久化）
        if self.opt.reuse_logs && is_last && compactions == 0 {
            // 复用当前日志文件（继续在原有文件后追加新操作）
            assert!(self.log.is_none());
            log!(self.opt.log, "reusing log file {:?}", filename);
            let oldsize = self.opt.env.size_of(Path::new(&filename))?;
            let oldfile = self.opt.env.open_appendable_file(Path::new(&filename))?;
            let lw = LogWriter::new_with_off(BufWriter::new(oldfile), oldsize);
            self.log = Some(lw);
            self.log_num = Some(log_num);
            self.mem = mem;
        } else if mem.len() > 0 {
            // Log is not reused, so write out the accumulated memtable.
            // 不复用日志，将剩余内存表内容持久化为SSTable
            save_manifest = true;
            self.write_l0_table(&mem, ve, None)?;
        }
        // save_manifest：是否需要保存元数据变更（如新增了 SSTable）。
        // max_seq：日志中记录的最大序列号（用于同步数据库的全局序列号）。
        Ok((save_manifest, max_seq))
    }

    /// delete_obsolete_files removes files that are no longer needed from the file system.
    ///
    /// 删除在文件系统中长时间不用的文件
    fn delete_obsolete_files(&mut self) -> Result<()> {
        // 当前所有活跃的文件
        let files = self.vset.borrow().live_files();
        // 获取该路径下所有的子文件和子文件夹
        let filenames = self.opt.env.children(Path::new(&self.path))?;
        for name in filenames {
            if let Ok((num, typ)) = parse_file_name(&name) {
                match typ {
                    FileType::Log => {
                        if num >= self.vset.borrow().log_num {
                            continue;
                        }
                    }
                    FileType::Descriptor => {
                        if num >= self.vset.borrow().manifest_num {
                            continue;
                        }
                    }
                    FileType::Table => {
                        if files.contains(&num) {
                            continue;
                        }
                    }
                    // NOTE: In this non-concurrent implementation, we likely never find temp
                    // files.
                    FileType::Temp => {
                        if files.contains(&num) {
                            continue;
                        }
                    }
                    FileType::Current | FileType::DBLock | FileType::InfoLog => continue,
                }

                // If we're here, delete this file.
                // 删除文件
                if typ == FileType::Table {
                    let _ = self.cache.borrow_mut().evict(num);
                }
                log!(self.opt.log, "Deleting file type={:?} num={}", typ, num);
                if let Err(e) = self.opt.env.delete(&self.path.join(&name)) {
                    log!(self.opt.log, "Deleting file num={} failed: {}", num, e);
                }
            }
        }
        Ok(())
    }

    /// acquire_lock acquires the lock file.
    fn acquire_lock(&mut self) -> Result<()> {
        let lock_r = self.opt.env.lock(Path::new(&lock_file_name(&self.path)));
        match lock_r {
            Ok(lockfile) => {
                self.lock = Some(lockfile);
                Ok(())
            }
            Err(ref e) if e.code == StatusCode::LockError => err(
                StatusCode::LockError,
                "database lock is held by another instance",
            ),
            Err(e) => Err(e),
        }
    }

    /// release_lock releases the lock file, if it's currently held.
    fn release_lock(&mut self) -> Result<()> {
        if let Some(l) = self.lock.take() {
            self.opt.env.unlock(l)
        } else {
            Ok(())
        }
    }

    /// Flush data to disk and release lock.
    pub fn close(&mut self) -> Result<()> {
        self.flush()?;
        self.release_lock()?;
        Ok(())
    }
}

impl DB {
    // WRITE //

    /// Adds a single entry. It's a short, non-synchronous, form of `write()`; in order to make
    /// sure that the written entry is on disk, call `flush()` afterwards.
    ///
    /// 这个 put 方法是数据库提供的单条键值对写入接口，用于将单个 (key, value) 键值对存入数据库。它本质上是对 write 方法的简化封装，提供了更便捷的单次写入方式。
    ///
    /// 这个 put 方法的核心价值是提供简洁的单条数据写入接口，主要作用包括：
    ///
    /// 简化单次写入操作
    ///
    /// 避免用户每次写入单条数据时都手动创建 WriteBatch，降低使用复杂度。
    ///
    /// 复用批量写入逻辑
    ///
    /// 通过封装 WriteBatch 和 write 方法，复用已有的批量写入逻辑（如内存表空间检查、日志记录、序列号管理等），保证代码一致性。
    ///
    /// 平衡性能与安全性
    ///
    /// 采用 sync: false 模式（默认不强制刷盘），优先保证写入性能；若用户需要确保数据持久化，可在调用 put 后手动调用 flush() 强制刷盘。
    pub fn put(&mut self, k: &[u8], v: &[u8]) -> Result<()> {
        // 初始化一个新的 WriteBatch（批量写入容器），用于包裹单个写入操作。
        let mut wb = WriteBatch::new();
        // 将当前要写入的键值对 (k, v) 添加到 WriteBatch 中。
        wb.put(k, v);
        // 调用之前分析的 write 方法，传入包含单条操作的 WriteBatch 和 sync: false（不强制同步刷盘）；
        // 最终会将该键值对写入内存表，并记录到预写日志（WAL），但不立即强制刷盘到物理磁盘。
        self.write(wb, false)
    }

    /// Deletes a single entry. Like with `put()`, you can call `flush()` to guarantee that
    /// the operation made it to disk.
    ///
    /// 这个 delete 方法是数据库提供的单条键删除接口，用于将指定的键从数据库中标记为删除。
    /// 它和 put 方法类似，是对批量写入接口 write 的简化封装，提供了便捷的单键删除方式。
    ///
    /// 这个 delete 方法的核心价值是提供简洁的单键删除接口，主要作用包括：
    ///
    /// 简化单次删除操作
    ///
    /// 避免用户手动创建 WriteBatch 来执行单个删除，降低使用复杂度。
    ///
    /// 复用批量删除逻辑
    ///
    /// 通过封装 WriteBatch 和 write 方法，复用已有的批量写入流程（如内存表操作、日志记录、序列号管理等），保证删除操作的一致性和安全性。
    ///
    /// 遵循 LSM 树的标记删除机制
    ///
    /// 本质是添加 “删除标记”（而非物理删除），符合 LSM 树 “顺序写优化” 的设计，后续通过压缩（compaction）过程完成真正的物理删除。
    pub fn delete(&mut self, k: &[u8]) -> Result<()> {
        // 初始化一个 WriteBatch（批量写入容器），用于包裹单个删除操作。
        let mut wb = WriteBatch::new();
        // 将当前要删除的键 k 添加到 WriteBatch 中，内部会记录一个 “删除标记”（而非直接物理删除数据）。
        wb.delete(k);
        // 调用 write 方法执行批量操作，传入包含单条删除操作的 WriteBatch 和 sync: false（不强制同步刷盘）；
        // 最终会将 “删除标记” 写入内存表，并记录到预写日志（WAL），但不立即强制刷盘到物理磁盘。
        self.write(wb, false)
    }

    /// Writes an entire WriteBatch. `sync` determines whether the write should be flushed to
    /// disk.
    ///
    /// 这个 write 方法是 LSM 树存储引擎中处理批量写入操作的核心接口，负责将 WriteBatch（批量写请求）中的操作应用到内存表，并同步到日志文件，
    /// 同时更新序列号确保数据一致性。它是数据库处理写入请求的关键入口，兼顾了性能和数据安全性。
    ///
    /// write 方法的核心价值在于实现高效、安全的批量写入，具体体现在：
    ///
    /// 批量处理优化：将多个写操作合并处理，减少内存表和日志的交互次数，提升写入性能；
    ///
    /// 数据安全保障：通过预写日志（WAL）确保即使进程崩溃，已提交的写操作也能从日志恢复；
    ///
    /// 版本控制：通过序列号管理数据版本，确保同一键的多次操作按顺序生效；
    ///
    /// 内存管理：自动处理内存表满的情况，通过 make_room_for_write 实现无缝切换，不阻塞写入。
    pub fn write(&mut self, batch: WriteBatch, sync: bool) -> Result<()> {
        assert!(self.log.is_some());
        // 本质是内存表的 “扩容 / 切换机制”，解决了 “内存表满了如何继续写入”
        // 调用 make_room_for_write 检查内存表（mem）是否有足够空间容纳新写入：
        // 若内存表未满，直接继续；
        // 若内存表已满，将其转换为 immutable 内存表（imm），并创建新的内存表接收新数据；
        // 这一步是 LSM 树 “内存表切换” 的核心，确保写入操作不被内存表大小限制。
        self.make_room_for_write(false)?;

        let entries = batch.count() as u64; // 批量操作中的条目数量
        let log = self.log.as_mut().unwrap(); // 日志文件句柄（预写日志 WAL）
        // 序列号（last_seq）用于标记操作的先后顺序，确保同一键的多次操作能按顺序生效（LSM 树通过序列号区分数据版本）。
        let next = self.vset.borrow().last_seq + 1; // 下一个可用的序列号
        // 写入到文件中
        // 将 WriteBatch 中的所有操作（插入 / 删除）应用到当前活跃的内存表（self.mem）；
        // 每个操作会被分配一个唯一的序列号（从 next 开始递增），确保版本正确性。
        // 这个是写入到内部跳表的
        batch.insert_into_memtable(next, &mut self.mem);
        // 预写日志（WAL） 是保障数据安全的关键：在数据写入内存表后，先写入日志文件，防止内存表数据因进程崩溃丢失；
        // sync: true 时，会调用 flush() 确保日志数据物理写入磁盘（牺牲性能换安全性）；sync: false 则依赖操作系统缓存异步刷盘（性能更好但有丢失风险）
        log.add_record(&batch.encode(next))?; // 将批量操作编码后的数据写入日志
        if sync {
            log.flush()?; // 若需要同步，强制将日志刷到磁盘
        }
        // 批量操作完成后，更新全局序列号（加上本次操作的条目数），确保后续操作使用新的序列号。
        self.vset.borrow_mut().last_seq += entries;
        Ok(())
    }

    /// flush makes sure that all pending changes (e.g. from put()) are stored on disk.
    pub fn flush(&mut self) -> Result<()> {
        if let Some(ref mut log) = self.log.as_mut() {
            log.flush()?;
        }
        Ok(())
    }
}

impl DB {
    // READ //

    /// 这个 get_internal 方法是 LSM 树（日志结构合并树）存储引擎中核心的数据查询接口，用于根据键（key）和序列号（seq）从多层存储结构中查找对应的值。
    /// 它会依次查询内存表、 immutable 内存表和磁盘上的 SSTable，并处理可能触发的压缩（compaction）操作，是数据库读取逻辑的核心实现。
    ///
    /// get_internal 方法的核心价值在于实现 LSM 树的多层级数据查询，确保：
    ///
    /// 数据可见性：优先查询内存中的最新数据（mem 和 imm），再查询磁盘上的历史数据（SSTable），符合 LSM 树 “新数据在内存，旧数据在磁盘” 的设计；
    ///
    /// 版本正确性：通过序列号（seq）过滤掉版本更新的数据，确保返回的是不晚于 seq 的最新版本；
    ///
    /// 自动维护：查询过程中检测并触发压缩操作，保持存储结构的高效性，避免查询性能随数据量增长而下降。
    fn get_internal(&mut self, seq: SequenceNumber, key: &[u8]) -> Result<Option<Bytes>> {
        // Using this lookup key will skip all entries with higher sequence numbers, because they
        // will compare "Lesser" using the InternalKeyCmp
        // LookupKey 是包含用户键（key）和序列号（seq）的复合键，用于在 LSM 树中精确定位最新版本的数据；
        // 序列号用于区分同一键的不同版本（LSM 树中，更新操作会生成新的版本，而非覆盖旧版本）。
        let lkey = LookupKey::new(key, seq);
        // self.mem 是当前活跃的内存表（通常是跳表结构），存储最新写入的数据；
        // 若内存表中找到键：
        // 若是普通值（Some(v)），直接返回该值；
        // 若是删除标记（None, true），返回 None（表示键已被删除）。
        // 最新数据在内存表（mem）
        match self.mem.get(&lkey) {
            (Some(v), _) => return Ok(Some(v)), // 找到值，直接返回
            // deleted entry
            (None, true) => return Ok(None), // 找到删除标记，返回 None
            // not found entry
            (None, false) => {} // 未找到，继续查询下一层
        }
        // self.imm 是已满的内存表（Immutable MemTable），已不能写入新数据，等待被压缩到磁盘；
        // 查询逻辑与内存表一致，确保覆盖所有内存中的数据版本。
        // 稍旧数据在 immutable 内存表（imm）
        if let Some(imm) = self.imm.as_ref() {
            match imm.get(&lkey) {
                (Some(v), _) => return Ok(Some(v)), // 找到值，直接返回
                // deleted entry
                (None, true) => return Ok(None), // 找到删除标记，返回 None
                // not found entry
                (None, false) => {} // 未找到，继续查询下一层
            }
        }

        // 最旧数据则持久化到磁盘的 SSTable 中。这段代码存在的核心原因是完整的数据查询必须覆盖所有存储层级，内存表中未找到的数据可能存在于磁盘的 SSTable 中。
        let mut do_compaction = false;
        let mut result = None;
        // Limiting the borrow scope of self.current.
        {
            // self.current() 表示磁盘上的当前 SSTable 层级（通常是多层级的 SSTable 集合）；
            // 通过 current.get 从 SSTable 中查询键，若找到则记录结果；
            // update_stats 用于更新访问统计，若满足压缩条件（如某个层级文件过多），标记 do_compaction = true。
            let current = self.current();
            let mut current = current.borrow_mut();
            // current.get(...) 从磁盘 SSTable 中查询数据
            // 这里 BlockIter 进
            if let Ok(Some((v, st))) = current.get(lkey.internal_key()) {
                // current.update_stats(st) 记录访问统计信息（如文件被访问的频率、层级文件数量等）
                if current.update_stats(st) {
                    // 若统计信息满足压缩条件（如某层级文件数过多），则标记 do_compaction = true，后续触发 maybe_do_compaction 进行后台压缩，合并小文件以优化查询性能
                    do_compaction = true; // 触发压缩标记
                }
                result = Some(v)
            }
        }
        // 若查询过程中标记了需要压缩，调用 maybe_do_compaction 执行后台压缩（合并 SSTable，减少文件数量，优化查询性能）；
        // 压缩操作失败时记录日志，但不影响当前查询结果。
        if do_compaction {
            if let Err(e) = self.maybe_do_compaction() {
                log!(self.opt.log, "error while doing compaction in get: {}", e);
            }
        }
        Ok(result)
    }

    /// get_at reads the value for a given key at or before snapshot. It returns Ok(None) if the
    /// entry wasn't found, and Err(_) if an error occurred.
    pub fn get_at(&mut self, snapshot: &Snapshot, key: &[u8]) -> Result<Option<Bytes>> {
        self.get_internal(snapshot.sequence(), key)
    }

    /// get is a simplified version of get_at(), translating errors to None.
    pub fn get(&mut self, key: &[u8]) -> Option<Bytes> {
        let seq = self.vset.borrow().last_seq;
        self.get_internal(seq, key).unwrap_or_default()
    }
}

impl DB {
    // ITERATOR //

    /// new_iter returns a DBIterator over the current state of the database. The iterator will not
    /// return elements added to the database after its creation.
    ///
    /// 这个 new_iter 方法是数据库提供的创建迭代器的接口，用于生成一个可以状态的迭代器（DBIterator），可以遍历数据库在 “当前时刻” 的所有键值对。
    /// 迭代器会固定在创建时的数据库状态，不会返回之后新增的数据，确保遍历结果的一致性。
    pub fn new_iter(&mut self) -> Result<DBIterator> {
        // 调用 get_snapshot 生成一个代表数据库当前状态的快照（记录当前最新序列号），确保迭代器的遍历范围被 “冻结” 在此时刻。
        let snapshot = self.get_snapshot();
        // 调用 new_iter_at 方法，传入刚创建的快照，生成一个基于该快照的迭代器；
        // 这个迭代器会根据快照的序列号过滤数据，只遍历该序列号之前存在的键值对。
        self.new_iter_at(snapshot)
    }

    /// new_iter_at returns a DBIterator at the supplied snapshot.
    ///
    /// 这段代码定义了 new_iter_at 方法，用于在指定的快照（Snapshot）上创建一个数据库迭代器（DBIterator），主要功能是提供对特定时间点的数据库数据进行遍历的能力。
    pub fn new_iter_at(&mut self, ss: Snapshot) -> Result<DBIterator> {
        Ok(DBIterator::new(
            self.opt.cmp.clone(), // 传入键比较器，确保迭代器按正确的键排序规则遍历数据
            self.vset.clone(), // 传入版本集合（包含所有 SSTable 层级信息），作为迭代器访问底层数据的基础
            self.merge_iterators()?,  // 生成合并后的底层数据迭代器（可能整合了内存表、Immutable 表和各层级 SSTable 的迭代器），确保能遍历所有相关数据
            ss,  // 指定的快照，迭代器将基于这个快照版本的数据进行遍历，不受后续写入操作影响
        ))
    }

    /// merge_iterators produces a MergingIter merging the entries in the memtable, the immutable
    /// memtable, and table files from all levels.
    ///
    /// 这个 merge_iterators 方法是 LSM 树存储引擎中用于创建合并迭代器（MergingIter） 的核心函数。
    /// 它会收集数据库中所有数据源（内存表、 immutable 内存表、各层级 SSTable 文件）的迭代器，然后将它们合并成一个统一的有序迭代器，供上层遍历使用。
    fn merge_iterators(&mut self) -> Result<MergingIter> {
        // 创建一个向量 iters，用于存储所有底层数据源的迭代器；
        // 类型 Box<dyn LdbIterator> 表示 “实现了 LdbIterator trait 的任意迭代器”，支持不同类型的数据源（内存或磁盘）。
        let mut iters: Vec<Box<dyn LdbIterator>> = vec![];
        // self.mem 是当前活跃的内存表（存储最新写入的数据）；
        // 若内存表非空（len() > 0），则创建其迭代器（self.mem.iter()），包装后加入 iters。
        if self.mem.len() > 0 {
            iters.push(Box::new(self.mem.iter()));
        }
        // self.imm 是 immutable 内存表（已写满、等待被压缩到磁盘的内存表）；
        // 若存在 immutable 内存表且非空，则创建其迭代器并加入 iters。
        if let Some(ref imm) = self.imm {
            if imm.len() > 0 {
                iters.push(Box::new(imm.iter()));
            }
        }

        // Add iterators for table files.
        // self.current() 获取数据库当前版本（Version），包含所有层级（Level 0 到 Level N）的 SSTable 元数据；
        // current.new_iters()? 为每个 SSTable 创建迭代器（用于读取磁盘上的键值对）；
        // 通过 extend 将这些 SSTable 迭代器加入 iters。
        // 获取当前版本的数据库
        let current = self.current();
        let current = current.borrow();
        // 从当前版本获取所有 SSTable 的迭代器，添加到集合中
        iters.extend(current.new_iters()?);
        // 使用数据库内部的比较器（self.internal_cmp，定义键的排序规则）和收集的所有迭代器（iters），创建 MergingIter 实例；
        // 返回合并迭代器，供上层（如 DBIterator）使用。
        Ok(MergingIter::new(self.internal_cmp.clone(), iters))
    }
}

impl DB {
    // SNAPSHOTS //

    /// Returns a snapshot at the current state. It can be used to retrieve entries from the
    /// database as they were at an earlier point in time.
    ///
    /// 这个 get_snapshot 方法是数据库对外提供的创建当前状态快照的接口，用于让用户获取数据库在 “此刻” 的一致性视图。
    /// 用户可以通过这个快照查询数据，确保看到的是调用该方法时的数据库状态，不受后续写入、删除操作的影响。
    ///
    /// get_snapshot 方法的核心价值是为用户提供 “时间点冻结” 能力，主要作用包括：
    ///
    /// 获取一致性视图:
    ///
    /// 快照创建时记录了当前最新的序列号（last_seq），后续查询时，数据库会确保只返回序列号小于等于该值的操作结果，忽略之后的所有修改，让用户看到的是 “创建快照瞬间” 的一致数据。
    ///
    /// 隔离读写操作:
    ///
    /// 即使其他线程或事务正在写入数据，通过该快照查询到的内容也不会受到影响，实现了 “读不加锁” 的高效并发访问（符合 LSM 树的多版本并发控制特性）。
    ///
    /// 支持历史查询:
    ///
    /// 用户可以保留该快照，在未来任意时间通过它查询 “过去某个时间点” 的数据，这对审计、备份、事务回滚等场景非常有用。
    pub fn get_snapshot(&mut self) -> Snapshot {
        // self.vset.borrow().last_seq：获取数据库当前的最新序列号（last_seq），这个序列号代表了数据库截至目前的所有操作版本（每一次写入 / 删除都会使序列号递增）；
        // self.snaps.new_snapshot(...)：调用快照列表（snaps）的 new_snapshot 方法，传入当前最新序列号，创建一个对应此序列号的新快照；
        // 最终返回创建的 Snapshot 实例，用户可通过该实例查询数据库在当前序列号时的状态。
        self.snaps.new_snapshot(self.vset.borrow().last_seq)
    }
}

impl DB {
    // STATISTICS //
    /// 这段代码定义了 add_stats 方法，用于将压缩操作（compaction）的统计信息添加到对应层级的统计数据中，是 LSM 树存储引擎中收集和累积压缩操作 metrics 的工具方法。
    ///
    /// CompactionStats 通常包含压缩操作的关键指标，如处理的数据量、耗时、删除的键数量、生成的新文件大小等。通过 add_stats 方法，存储引擎可以按层级累积这些统计信息，用于：
    ///
    /// 监控各层级的压缩效率和负载
    ///
    /// 优化压缩策略（如动态调整触发阈值）
    ///
    /// 提供性能分析和调试依据
    // 简单来说，它是压缩统计信息的 “层级累加器”，负责按层级收集压缩操作的各项指标，为存储引擎的监控和优化提供数据支持。
    fn add_stats(&mut self, level: usize, cs: CompactionStats) {
        // 参数校验：assert!(level < NUM_LEVELS) 确保传入的层级（level）是有效的（在预设的层级数量范围内）。
        assert!(level < NUM_LEVELS);
        // 累积统计信息：将压缩统计数据（cs: CompactionStats）添加到当前对象的 cstats 数组中对应层级（level）的统计项中（通过 add 方法合并）。
        self.cstats[level].add(cs);
    }

    /// Trigger a compaction based on where this key is located in the different levels.
    fn record_read_sample(&mut self, k: InternalKey<'_>) {
        let current = self.current();
        if current.borrow_mut().record_read_sample(k) {
            if let Err(e) = self.maybe_do_compaction() {
                log!(self.opt.log, "record_read_sample: compaction failed: {}", e);
            }
        }
    }
}

impl DB {
    // COMPACTIONS //

    /// make_room_for_write checks if the memtable has become too large, and triggers a compaction
    /// if it's the case.
    ///
    /// 这个 make_room_for_write 方法是数据库为新写入操作腾出内存空间的核心逻辑，主要用于处理内存表（MemTable）满了的场景，确保后续写入能正常执行。
    ///
    /// 本质是内存表的 “扩容 / 切换机制”，解决了 “内存表满了如何继续写入”
    fn make_room_for_write(&mut self, force: bool) -> Result<()> {
        // 不是强制触发（!force），且当前内存表（self.mem）的内存占用小于配置的写入缓冲区大小（write_buffer_size）；
        // 或内存表为空（self.mem.len() == 0）。
        if !force && self.mem.approx_mem_usage() < self.opt.write_buffer_size || self.mem.len() == 0 {
            Ok(())
        } else {
            // 内存不足时：创建新内存表，旧表转为不可变状态
            // Create new memtable. 获取一个新的编号
            let logn = self.vset.borrow_mut().new_file_number();
            // 打开文件
            let logf = self.opt.env.open_writable_file(Path::new(&log_file_name(&self.path, logn)));
            if logf.is_err() {
                // 回收账号
                self.vset.borrow_mut().reuse_file_number(logn);
                Err(logf.err().unwrap())  // 返回错误
            } else {
                self.log = Some(LogWriter::new(BufWriter::new(logf.unwrap())));
                self.log_num = Some(logn);  // 记录当前 WAL 日志的文件编号

                let mut imm = MemTable::new(self.opt.cmp.clone());  // 新建空的内存表
                mem::swap(&mut imm, &mut self.mem);  // 交换：旧表放入 `imm`，新空表成为 `self.mem`
                self.imm = Some(imm); // 旧表标记为“不可变内存表”（后续会被压缩到磁盘）
                // 触发压缩：将不可变内存表落地到磁盘
                self.maybe_do_compaction()
            }
        }
    }

    /// maybe_do_compaction starts a blocking compaction if it makes sense.
    ///
    /// 这个 maybe_do_compaction 方法是数据库的压缩调度入口，用于判断并触发必要的压缩操作，以保持数据库的读写性能和存储效率。其核心逻辑可以总结为：
    fn maybe_do_compaction(&mut self) -> Result<()> {
        // 优先处理不可变内存表（imm）
        if self.imm.is_some() {
            // 如果存在 imm（即内存表已满并转为不可变状态），则调用 compact_memtable() 将其持久化为 L0 层的 SSTable 文件。这一步是为了避免内存占用过高，并将内存数据固化到磁盘。
            self.compact_memtable()?;
        }
        // Issue #34 PR #36: after compacting a memtable into an L0 file, it is possible that the L0 files need to be merged and promoted.
        // 通过 vset.borrow().needs_compaction() 判断是否需要对现有 SSTable 层级（如 L0 到 Ln）进行压缩。
        if self.vset.borrow().needs_compaction() {
            // 若需要，调用 pick_compaction() 选择最合适的压缩任务（例如文件数量过多的层级或大小超标的层级）。
            let c = self.vset.borrow_mut().pick_compaction();
            if let Some(c) = c {
                // 若存在合适的压缩任务，则通过 start_compaction(c) 启动压缩过程（合并文件、清理过期数据、提升到更高层级等）。
                self.start_compaction(c)
            } else {
                Ok(())
            }
        } else {
            Ok(())
        }
    }

    /// compact_range triggers an immediate compaction on the specified key range. Repeatedly
    /// calling this without actually adding new keys is not useful.
    ///
    /// Compactions in general will cause the database to find entries more quickly, and take up
    /// less space on disk.
    ///
    /// 这个 compact_range 方法是 LSM 树存储引擎中用于触发指定键范围的手动压缩操作的核心接口。
    /// 它可以强制对数据库中特定键范围内的数据进行合并与清理，优化后续查询性能并减少磁盘占用。
    ///
    /// compact_range 的核心价值是主动优化特定键范围的存储结构，主要作用包括：
    ///
    /// 清理无效数据
    ///
    /// 压缩过程中会删除被标记删除的键、覆盖旧版本的键，减少磁盘存储的无效数据，节省空间。
    ///
    /// 提升查询性能
    ///
    /// 合并重叠的 SSTable 文件后，查询指定范围的键时需要访问的文件数量减少，降低 I/O 开销，加速查询。
    ///
    /// 控制压缩粒度
    ///
    /// 相比后台自动压缩，手动调用 compact_range 可以针对热点范围（如频繁读写的键区间）进行定向优化，避免自动压缩的延迟。
    ///
    /// 维护数据结构稳定性
    ///
    /// 强制压缩可以防止特定层级的文件数量过多或范围过度重叠，维持 LSM 树的平衡结构。
    pub fn compact_range(&mut self, from: &[u8], to: &[u8]) -> Result<()> {
        // 目的：找到与目标键范围 [from, to] 重叠的最高层级（max_level），后续需对从 Level 0 到该层级的数据进行压缩；
        // 原理：LSM 树中高层级（如 Level 2+）的数据范围更稳定，若目标范围与某层级重叠，说明需要从低层到高层逐步压缩以彻底清理该范围。
        let mut max_level = 1;
        {
            let v = self.vset.borrow().current(); // 获取当前数据库版本
            let v = v.borrow();
            // 检查各层级是否有与 [from, to] 范围重叠的数据
            for l in 1..NUM_LEVELS - 1 {
                if v.overlap_in_level(l, from, to) {
                    max_level = l;  // 更新最大层级为有重叠的最高层级
                }
            }
        }

        // Compact memtable.
        // 强制将内存表数据写入 immutable 内存表（为新写入腾出空间，同时确保内存数据参与压缩）
        // make_room_for_write(true) 会触发内存表的刷盘准备：若当前内存表非空，将其转换为 immutable 内存表（等待压缩到磁盘）；
        // 这一步确保内存中与目标范围相关的最新数据被纳入压缩流程，避免遗漏。
        self.make_room_for_write(true)?;
        // 构建起始内部键（包含最大序列号，确保覆盖所有版本）
        // LSM 树内部键包含用户键 + 序列号 + 操作类型，这里将用户传入的 from 和 to 转换为内部键格式：
        // ifrom 用最大序列号（MAX_SEQUENCE_NUMBER）确保覆盖该键的所有历史版本；
        // iend 用删除标记确保压缩范围包含 to 键本身。
        let mut ifrom = LookupKey::new(from, MAX_SEQUENCE_NUMBER).internal_key().to_vec();
        // 构建结束内部键（删除标记，确保覆盖到 to 键的所有版本）
        let iend = LookupKey::new_full(to, 0, ValueType::TypeDeletion);
        // 从 Level 0 到 max_level 逐层压缩
        // 逐层压缩：对每个层级 l 循环执行压缩，直到该层级不再有与目标范围重叠的文件；
        // 压缩任务创建：vset.compact_range(...) 分析当前层级，找到与 [ifrom, iend] 重叠的 SSTable 文件，生成压缩任务（c）；
        // 动态调整范围：每次压缩后，将 ifrom 更新为本次压缩输入文件的最大键，避免重复处理已压缩的范围；
        // 执行压缩：start_compaction(c) 实际执行压缩（合并重叠文件、清理无效数据、下沉到下一层级）。
        for l in 0..max_level + 1 {
            loop {
                // 尝试在当前层级 l 上创建针对 [ifrom, iend] 的压缩任务
                // 这个 c_ 里面有需要压缩的文件列表
                let c_ = self
                    .vset
                    .borrow_mut()
                    .compact_range(l, &ifrom, iend.internal_key());
                if let Some(c) = c_ {
                    // Update ifrom to the largest key of the last file in this compaction.
                    // 更新起始键为本次压缩输入文件的最大键（避免重复压缩）
                    let ix = c.num_inputs(0) - 1;
                    ifrom.clone_from(&c.input(0, ix).largest.into());
                    // 执行压缩任务
                    // 这个 c 里面有需要压缩的文件列表
                    self.start_compaction(c)?;
                } else {
                    // 无更多可压缩的文件，退出当前层级循环
                    break;
                }
            }
        }
        Ok(())
    }

    /// start_compaction dispatches the different kinds of compactions depending on the current
    /// state of the database.
    ///
    /// 这个 start_compaction 方法是 LSM 树存储引擎中用于启动并执行压缩任务（Compaction）的核心调度函数。
    /// 它根据压缩任务的类型（简单移动或常规压缩），选择不同的执行路径，最终完成数据的合并、迁移或清理，并更新数据库状态。
    ///
    /// 核心本质：常规压缩需要合并源层级和目标层级的重叠文件，清理过期 / 删除的键值对（基于序列号），并生成新的 SSTable 文件；
    ///
    /// 关键步骤解析：
    ///
    /// smallest 序列号：确保压缩后的数据不会删除快照中仍需可见的版本（多版本并发控制的核心）；
    ///
    /// do_compaction_work：实际执行键值对合并、去重、排序和写入新文件的逻辑；
    ///
    /// install_compaction_results：将新文件添加到数据库版本，标记旧文件为待删除；
    ///
    /// delete_obsolete_files：物理删除已被替代的旧文件，释放磁盘空间。
    fn start_compaction(&mut self, mut compaction: Compaction) -> Result<()> {
        if compaction.is_trivial_move() {
            // 断言输入文件数量为1（简单移动仅处理单个文件）
            assert_eq!(1, compaction.num_inputs(0));
            let f = compaction.input(0, 0); // 获取待移动的文件
            let num = f.num;   // 文件编号
            let size = f.size;  // 文件大小
            let level = compaction.level(); // 当前层级

            // 构建版本变更指令：
            // 1. 从源层级删除该文件
            compaction.edit().delete_file(level, num);
            // 2. 将该文件添加到下一层级（level+1）
            compaction.edit().add_file(level + 1, f);

            // 应用变更并持久化：
            // - 将变更写入 MANIFEST 日志
            // - 生成新的数据库版本并设为当前版本
            let r = self.vset.borrow_mut().log_and_apply(compaction.into_edit());
            if let Err(e) = r {
                log!(self.opt.log, "trivial move failed: {}", e);
                Err(e)
            } else {
                // 记录移动成功的日志，包含文件信息和层级变化
                log!(self.opt.log,"Moved num={} bytes={} from L{} to L{}",num,size,level,level + 1);
                log!(self.opt.log,"Summary: {}",self.vset.borrow().current_summary());
                Ok(())
            }
        } else {
            // 步骤1：确定压缩时需要保留的最小序列号（用于清理过期数据）
            let smallest = if self.snaps.empty() {
                // 若无快照，保留到数据库最新序列号
                self.vset.borrow().last_seq
            } else {
                // 若有快照，保留到最旧快照的序列号（确保快照可见性）
                self.snaps.oldest()
            };
            // 步骤2：创建压缩状态对象，初始化压缩上下文
            // compaction 里面有要压缩的文件
            let mut state = CompactionState::new(compaction, smallest);
            // 步骤3：执行实际的压缩工作（核心逻辑）
            if let Err(e) = self.do_compaction_work(&mut state) {
                // 压缩失败时清理临时文件
                state.cleanup(self.opt.env.as_ref().as_ref(), &self.path);
                log!(self.opt.log, "Compaction work failed: {}", e);
            }
            // 步骤4：安装压缩结果（更新数据库版本）
            self.install_compaction_results(state)?;
            // 步骤5：记录压缩完成日志，包含当前版本摘要
            log!(self.opt.log,"Compaction finished: {}",self.vset.borrow().current_summary());
            // 步骤6：删除不再需要的旧文件（已被新文件替代的文件）
            self.delete_obsolete_files()
        }
    }

    /// 这段代码定义了 compact_memtable 方法，用于将 LSM 树存储引擎中的不可变内存表（immutable memtable，简称 imm）持久化到磁盘，生成新的 SSTable 并更新版本信息，是内存数据落地的关键操作。
    ///
    /// 这个方法的核心作用是完成内存数据到磁盘的持久化，是 LSM 树 “内存表 -> 磁盘表” 转换的核心步骤。当内存表满后，会转为不可变内存表，再通过该方法写入磁盘成为 L0 层的 SSTable，
    /// 确保数据不会因内存问题丢失，并为后续压缩操作提供基础。
    ///
    /// 简单来说，它是内存数据的 “落地器”，负责将不可变内存表中的数据安全写入磁盘并更新存储引擎的版本信息。
    fn compact_memtable(&mut self) -> Result<()> {
        // assert!(self.imm.is_some()) 确保存在可持久化的不可变内存表（imm），避免无数据时执行操作。
        assert!(self.imm.is_some());
        // 创建 VersionEdit 实例（ve），用于记录本次操作对版本的修改（如添加新 SSTable）。
        let mut ve = VersionEdit::new();
        // 获取当前版本（base）的引用，作为版本更新的基础。
        let base = self.current();
        // 通过 self.imm.take().unwrap() 取出不可变内存表（imm），后续若操作失败会将其放回（self.imm = Some(imm)）。
        let imm = self.imm.take().unwrap();
        // 调用 write_l0_table 将内存表数据写入磁盘，生成 L0 层的 SSTable，并将新文件信息记录到 ve 中。
        if let Err(e) = self.write_l0_table(&imm, &mut ve, Some(&base.borrow())) {
            // 若写入失败，返回错误并恢复 imm 状态。
            self.imm = Some(imm);
            return Err(e);
        }
        // 设置日志编号（ve.set_log_num），关联本次操作与 WAL（Write-Ahead Log）。
        ve.set_log_num(self.log_num.unwrap_or(0));
        // 通过 vset.borrow_mut().log_and_apply(ve) 将版本编辑应用到版本集合（vset），生成新的版本，使新 SSTable 生效。
        self.vset.borrow_mut().log_and_apply(ve)?;
        // 调用 delete_obsolete_files 删除不再需要的旧文件（如已合并的 SSTable 或过期的日志文件），释放磁盘空间。
        // 即使清理失败，也仅记录日志而不影响主流程。
        if let Err(e) = self.delete_obsolete_files() {
            log!(self.opt.log, "Error deleting obsolete files: {}", e);
        }
        Ok(())
    }

    /// write_l0_table writes the given memtable to a table file.
    ///
    /// 这个 write_l0_table 方法是数据库中将内存表（MemTable）的数据持久化为 L0 层 SSTable 文件的核心逻辑，
    /// 同时会更新版本变更信息和统计数据。它是内存数据落地到磁盘的关键执行步骤。
    ///
    /// 本质是将内存表中的数据安全落地为磁盘 SSTable 文件，并完成相关的元数据、缓存、统计等配套操作，具体包括：
    ///
    /// 生成唯一标识的 SSTable 文件；
    ///
    /// 验证文件有效性并接入缓存系统；
    ///
    /// 记录写入性能指标；
    ///
    /// 为版本更新提供 “新增文件” 的变更信息。
    fn write_l0_table(
        &mut self,
        memt: &MemTable,
        ve: &mut VersionEdit,
        base: Option<&Version>,
    ) -> Result<()> {
        // 返回一个时间
        let start_ts = self.opt.env.micros();
        // 获取一个新的文件编号
        let num = self.vset.borrow_mut().new_file_number();
        log!(self.opt.log, "Start write of L0 table {:06}", num);
        // 调用 build_table 方法（之前解析过的核心工具函数），将内存表（memt）的键值对通过迭代器遍历，按 SSTable 格式写入磁盘文件。
        let fmd = build_table(&self.path, &self.opt, memt.iter(), num)?;
        log!(self.opt.log, "L0 table {:06} has {} bytes", num, fmd.size);

        // Wrote empty table.
        // 若生成的文件为空（无键值对），则回收文件编号（reuse_file_number），避免资源浪费，直接返回成功。
        if fmd.size == 0 {
            self.vset.borrow_mut().reuse_file_number(num);
            return Ok(());
        }

        // 尝试从 TableCache 中获取刚创建的文件
        let cache_result = self.cache.borrow_mut().get_table(num);
        if let Err(e) = cache_result {
            log!(self.opt.log,"L0 table {:06} not returned by cache: {}",num,e);
            // 若失败（如文件未正确写入或缓存异常），则删除磁盘上的无效文件，返回错误。
            // 这一步确保新文件能被缓存系统正常识别，为后续查询做准备。
            let _ = self.opt.env.delete(Path::new(&table_file_name(&self.path, num)));
            return Err(e);
        }

        let stats = CompactionStats {
            micros: self.opt.env.micros() - start_ts, // 写入耗时（微秒）
            written: fmd.size, // 写入字节数
            ..Default::default()
        };

        let mut level = 0;
        // 确定文件最终落地的层级
        if let Some(b) = base {
            level = b.pick_memtable_output_level(
                parse_internal_key(&fmd.smallest).2, // 最小键序列号
                parse_internal_key(&fmd.largest).2, // 最大键序列号
            );
        }

        self.add_stats(level, stats); // 将统计信息添加到数据库全局统计中
        ve.add_file(level, fmd);  // 将新文件的元数据添加到版本变更指令（VersionEdit）中

        Ok(())
    }
    /// 这个 do_compaction_work 方法是数据库压缩流程的核心执行环节，负责完成压缩任务的具体工作：读取待压缩的输入文件、合并键值对、过滤过期数据、生成新的 SSTable 文件，并记录压缩过程的统计信息。
    fn do_compaction_work(&mut self, cs: &mut CompactionState) -> Result<()> {
        // 确认待压缩层级有文件，且压缩输出构建器（builder）未初始化，确保压缩流程从干净状态开始。
        {
            let current = self.vset.borrow().current();
            assert!(current.borrow().num_level_files(cs.compaction.level()) > 0);
            assert!(cs.builder.is_none());
        }
        let start_ts = self.opt.env.micros();
        log!(
            self.opt.log,
            "Compacting {} files at L{} and {} files at L{}",
            cs.compaction.num_inputs(0),
            cs.compaction.level(),
            cs.compaction.num_inputs(1),
            cs.compaction.level() + 1
        );
        // 创建输入迭代器：通过 make_input_iterator 生成一个统一的迭代器，可按键序遍历所有待压缩文件的键值对（屏蔽多文件的底层差异）。
        // 这里是 MergingIter
        let mut input = self.vset.borrow().make_input_iterator(&cs.compaction);
        input.seek_to_first();

        let mut last_seq_for_key = MAX_SEQUENCE_NUMBER;

        let mut have_ukey = false;
        let mut current_ukey = vec![];
        // 循环遍历输入迭代器的所有键值对，核心逻辑围绕 “去重、过滤过期数据、生成新文件” 展开
        // 这里是 MergingIter
        while input.valid() {
            // TODO: Do we need to do a memtable compaction here? Probably not, in the sequential case.
            let (key_bytes, val_bytes) = input.current().expect("Iterator should be valid here");
            // 它是压缩过程中的 “文件拆分控制器”，当检测到当前输出文件与更高层级文件的重叠数据过多时，会触发新文件的创建，以保持存储结构的高效性。
            if cs.compaction.should_stop_before(&key_bytes) && cs.builder.is_some() {
                // 负责结束 SSTable 文件的构建、验证文件可用性，并记录关键信息，确保压缩生成的新文件有效且可被数据库后续使用
                self.finish_compaction_output(cs)?;
            }
            // 解析内部键
            let (ktyp, seq, ukey) = parse_internal_key(&key_bytes);
            if seq == 0 {
                // Parsing failed.
                log!(self.opt.log, "Encountered seq=0 in key: {:?}", &key_bytes);
                last_seq_for_key = MAX_SEQUENCE_NUMBER;
                have_ukey = false;
                current_ukey.clear();
                // 这里是 MergingIter
                input.advance();
                continue;
            }

            if !have_ukey || self.opt.cmp.cmp(ukey, &current_ukey) != Ordering::Equal {
                // First occurrence of this key.
                // 遇到新的用户键，重置该键的状态（记录当前键、重置最新序列号）
                current_ukey.clear();
                current_ukey.extend_from_slice(ukey);
                have_ukey = true;
                last_seq_for_key = MAX_SEQUENCE_NUMBER;
            }

            // We can omit the key under the following conditions:
            // 过滤条件1：当前键的序列号早于所有快照的最小序列号（数据已过期，可丢弃）
            if last_seq_for_key <= cs.smallest_seq {
                last_seq_for_key = seq;
                // 这里是 MergingIter
                input.advance();
                continue;
            }
            // Entry is deletion; no older version is observable by any snapshot; and all entries in compacted levels with smaller sequence numbers will
            // 过滤条件2：删除标记 + 序列号过期 + 是基础层级（彻底删除，无需保留）
            if ktyp == ValueType::TypeDeletion && seq <= cs.smallest_seq && cs.compaction.is_base_level_for(ukey) {
                last_seq_for_key = seq;
                // 这里是 MergingIter
                input.advance();
                continue;
            }

            last_seq_for_key = seq;

            if cs.builder.is_none() {
                let fnum = self.vset.borrow_mut().new_file_number(); // 申请新文件编号
                let fmd = FileMetaData {
                    num: fnum,
                    ..Default::default()
                };

                let fname = table_file_name(&self.path, fnum); // 生成新文件路径
                let f = self.opt.env.open_writable_file(Path::new(&fname))?; // 打开写入文件
                let f = Box::new(BufWriter::new(f));
                // 创建 SSTable 构建器（包装缓冲写入流）
                cs.builder = Some(TableBuilder::new(self.opt.clone(), f));
                cs.outputs.push(fmd); // 记录新文件元数据
            }
            // 记录新文件的最小/最大键（用于元数据）
            if cs.builder.as_ref().unwrap().entries() == 0 {
                cs.current_output().smallest.clone_from(&key_bytes);
            }
            cs.current_output().largest.clone_from(&key_bytes);
            // 写入键值对到新文件
            cs.builder.as_mut().unwrap().add(&key_bytes, &val_bytes)?;
            // NOTE: Adjust max file size based on level.
            // 若当前文件大小超过阈值，结束当前文件，后续数据写入新文件
            if cs.builder.as_ref().unwrap().size_estimate() > self.opt.max_file_size {
                self.finish_compaction_output(cs)?;
            }
            // 这里是 MergingIter
            input.advance();
        }

        // 完成最后一个输出文件：循环结束后，若仍有活跃的构建器，调用 finish_compaction_output 完成文件写入并更新元数据（如文件大小）。
        if cs.builder.is_some() {
            self.finish_compaction_output(cs)?;
        }
        // 统计压缩性能：计算压缩耗时（micros）、读取的数据量（read，输入文件总大小）、写入的数据量（written，输出文件总大小），并更新全局压缩统计（cstats）。
        let mut stats = CompactionStats {
            micros: self.opt.env.micros() - start_ts,
            ..Default::default()
        };
        for parent in 0..2 {
            for inp in 0..cs.compaction.num_inputs(parent) {
                stats.read += cs.compaction.input(parent, inp).size;
            }
        }
        for output in &cs.outputs {
            stats.written += output.size;
        }
        self.cstats[cs.compaction.level()].add(stats);
        Ok(())
    }

    /// 这个 finish_compaction_output 方法是 LSM 树压缩过程中用于完成单个压缩输出文件的最终处理的核心函数。
    /// 它负责结束 SSTable 文件的构建、验证文件可用性，并记录关键信息，确保压缩生成的新文件有效且可被数据库后续使用。
    ///
    /// finish_compaction_output 的核心价值是确保压缩生成的 SSTable 文件有效且可被数据库使用，主要作用包括：
    ///
    /// 1.完成文件构建流程
    ///
    /// 调用 TableBuilder::finish() 完成 SSTable 的最终写入和磁盘持久化，将内存中的临时数据转换为可长期使用的磁盘文件。
    ///
    /// 2.验证文件有效性
    ///
    /// 通过表缓存加载新文件并检查是否可读取，提前发现文件损坏或构建错误，避免后续查询操作因无效文件而失败。
    ///
    /// 3.更新压缩状态元数据
    ///
    /// 记录文件大小、累加总输出字节数，为后续的版本更新（如 install_compaction_results）提供准确的元数据。
    ///
    /// 4.提供可审计的日志
    ///
    /// 记录新文件的关键信息，便于追踪压缩效果（如每个文件的键数量、大小）和排查问题。
    fn finish_compaction_output(&mut self, cs: &mut CompactionState) -> Result<()> {
        // cs.builder 是用于构建 SSTable 文件的构建器（如 TableBuilder），必须已初始化才能执行后续操作；
        // output_num 是压缩生成的新 SSTable 的唯一编号，用于标识文件（如文件名通常包含该编号），必须有效。
        assert!(cs.builder.is_some()); // 确保 SSTable 构建器已初始化
        let output_num = cs.current_output().num; // 确保 SSTable 构建器已初始化
        assert!(output_num > 0); // 确保文件编号有效（大于 0）

        // The original checks if the input iterator has an OK status. For this, we'd need to
        // extend the LdbIterator interface though -- let's see if we can without for now.
        // (it's not good for corruptions, in any case)
        // 取出构建器（take() 会转移所有权，避免后续重复使用）
        let b = cs.builder.take().unwrap();
        let entries = b.entries();  // 获取该 SSTable 包含的键值对数量
        // b.finish() 是 SSTable 构建的关键步骤：将内存中的键值对、索引等数据写入磁盘文件，完成校验和计算，并关闭文件；
        // 记录键值对数量（entries）和文件大小（bytes），用于元数据更新和日志记录。
        let bytes = b.finish()?;   // 完成文件构建（写入磁盘、生成索引等），返回文件大小（字节数）
        cs.total_bytes += bytes;    // 累加当前压缩任务的总输出字节数
        // 累加当前压缩任务的总输出字节数
        cs.current_output().size = bytes;

        if entries > 0 {  // 仅当文件包含键值对时才需要验证
            // Verify that table can be used. (Separating get_table() because borrowing in an if
            // let expression is dangerous).
            // 尝试从表缓存中加载刚创建的文件，验证其可读取性
            let r = self.cache.borrow_mut().get_table(output_num);
            if let Err(e) = r {
                log!(self.opt.log, "New table can't be read: {}", e);
                return Err(e);  // 验证失败，返回错误
            }
            log!(self.opt.log,"New table num={}: keys={} size={}",output_num,entries,bytes);
        }
        Ok(())
    }

    /// 这个 install_compaction_results 方法是 LSM 树存储引擎中用于将压缩操作（Compaction）的结果应用到数据库版本中的核心函数。
    /// 它负责记录压缩压缩生成的新文件、记录变更、更新数据库版本，确保压缩后的结果被正确持久化并反映到后续的读写操作中。
    fn install_compaction_results(&mut self, mut cs: CompactionState) -> Result<()> {
        // 打印压缩操作的关键信息到日志，包括参与压缩的源文件数、层级、目标层级、生成文件总大小等，用于调试和监控压缩过程。
        log!(
            self.opt.log,
            "Compacted {} L{} files + {} L{} files => {}B",
            cs.compaction.num_inputs(0),  // 源层级（level）参与压缩的文件数
            cs.compaction.level(),   // 源层级编号
            cs.compaction.num_inputs(1),  // 目标层级（level+1）参与压缩的文件数
            cs.compaction.level() + 1,    // 目标层级编号
            cs.total_bytes      // 压缩后生成文件的总字节数
        );
        // add_input_deletions 方法会在压缩编辑（VersionEdit）中记录需要删除的源文件（包括源层级 level 和目标层级 level+1 中参与压缩的文件），确保这些旧文件在新版本中被移除。
        cs.compaction.add_input_deletions();  // 标记源文件为待删除
        let level = cs.compaction.level();   // 源层级编号
        // cs.outputs 存储压缩过程中生成的新 SSTable 文件（已合并、去重、排序后的结果）；
        // 通过 add_file 方法将这些新文件记录到压缩编辑中，指定它们属于目标层级 level+1（LSM 树中压缩结果通常下沉到下一层级）。
        for output in &cs.outputs {
            // 将压缩生成的新文件添加到目标层级（level+1）
            cs.compaction.edit().add_file(level + 1, output.clone());
        }
        // 将 VersionEdit 写入 MANIFEST 日志（持久化版本变更历史）；
        // 基于当前版本和 VersionEdit 生成新的数据库版本；
        // 将新版本设置为当前活跃版本，确保后续读写操作使用最新的文件集合。
        self.vset
            .borrow_mut()
            .log_and_apply(cs.compaction.into_edit())   // 将压缩编辑（包含删除旧文件、添加新文件的操作）应用到版本集中
    }
}

impl Drop for DB {
    fn drop(&mut self) {
        self.flush().ok();
        let _ = self.release_lock();
    }
}

/// 这段代码定义了 CompactionState 结构体，用于跟踪和管理 LSM 树存储引擎中压缩操作（compaction）的状态信息，是压缩过程中的核心状态容器。
///
/// 在 LSM 树中，压缩操作的主要目的是合并多个 SSTable（可能来自同一层级或不同层级），去除重复或过期的键值对，并生成新的、更有序的 SSTable。CompactionState 正是用于记录这一过程中的关键信息。
///
/// CompactionState 的核心作用是在压缩过程中集中管理所有动态状态，包括输入任务、输出结果、临时构建器和统计信息等。
/// 通过这个结构体，压缩逻辑可以清晰地跟踪进度、管理输出文件的创建，并最终完成旧数据的合并和新数据的生成。
struct CompactionState {
    /// 存储当前压缩任务的元信息（如参与压缩的 SSTable 列表、源层级、目标层级等），是整个压缩操作的基础参数。
    compaction: Compaction,
    /// 记录当前压缩过程中遇到的最小序列编号（SequenceNumber）。序列编号用于标识键的版本（越大越新），跟踪最小序列编号有助于判断数据的时效性，或用于后续的垃圾回收等操作。
    smallest_seq: SequenceNumber,
    /// 存储压缩过程中生成的新 SSTable 的元数据列表（如文件编号、键范围、大小等）。这些新文件最终会被添加到目标层级，替代参与压缩的旧文件。
    outputs: Vec<FileMetaData>,
    /// 可选的 SSTable 构造器（TableBuilder），用于构建新的输出 SSTable。当当前构建的 SSTable 达到预设大小时，会被写入磁盘并创建新的 TableBuilder，因此该字段可能为 None（表示等待创建新构建器）。
    builder: Option<TableBuilder<Box<dyn Write>>>,
    /// 累计当前压缩操作已处理的总数据量（字节数），用于监控压缩进度或判断是否需要拆分输出文件。
    total_bytes: usize,
}

impl CompactionState {
    fn new(c: Compaction, smallest: SequenceNumber) -> CompactionState {
        CompactionState {
            compaction: c,
            smallest_seq: smallest,
            outputs: vec![],
            builder: None,
            total_bytes: 0,
        }
    }

    /// 这段代码定义了 current_output 方法，用于获取压缩状态（CompactionState）中当前正在构建的输出 SSTable 的元数据（FileMetaData），是压缩过程中访问和更新当前输出文件信息的便捷接口。
    ///
    /// 在压缩过程中，当需要向当前输出 SSTable 写入数据时，通常需要不断更新其元数据（例如，随着键值对的添加，更新文件的最大键、总大小等信息）。
    /// current_output 方法提供了直接访问当前输出文件元数据的途径，确保这些信息能被及时更新。
    ///
    /// 简单来说，它是压缩过程中 “当前输出文件的元数据访问器”，用于获取并修改正在构建的 SSTable 的详细信息。
    fn current_output(&mut self) -> &mut FileMetaData {
        // 首先通过 self.outputs.len() 获取当前已生成的输出文件列表（outputs）的长度。
        let len = self.outputs.len();
        // 然后通过索引 len - 1 访问列表中的最后一个元素，即当前正在构建的输出文件的元数据。
        // 返回该元数据的可变引用（&mut FileMetaData），允许调用方修改其属性（如更新键范围、文件大小等）。
        &mut self.outputs[len - 1]
    }

    /// cleanup cleans up after an aborted compaction.
    fn cleanup<P: AsRef<Path>>(&mut self, env: &dyn Env, name: P) {
        for o in self.outputs.drain(..) {
            let name = table_file_name(name.as_ref(), o.num);
            let _ = env.delete(&name);
        }
    }
}

#[derive(Debug, Default)]
struct CompactionStats {
    micros: u64,
    read: usize,
    written: usize,
}

impl CompactionStats {
    fn add(&mut self, cs: CompactionStats) {
        self.micros += cs.micros;
        self.read += cs.read;
        self.written += cs.written;
    }
}

/// 这个 build_table 方法用于将内存中的键值对数据持久化为磁盘上的 SSTable 文件，并生成该文件的元数据（用于后续的版本管理和查询）。
/// 它是内存数据（如 MemTable）落地到磁盘的关键步骤。
pub fn build_table<I: LdbIterator, P: AsRef<Path>>(
    dbname: P, //数据库路径（dbname）
    opt: &Options, //配置（opt）
    mut from: I, //键值对迭代器（from，通常来自内存表）
    num: FileNum, //文件编号（num，唯一标识新文件）。
) -> Result<FileMetaData> {
    from.reset();
    // 传一个 path_name, 后缀加上 ldb
    let filename = table_file_name(dbname.as_ref(), num);

    let (mut kbuf, mut vbuf) = (vec![], vec![]);
    let mut firstkey = None;
    // lastkey is what remains in kbuf.

    // Clean up file if write fails at any point.
    //
    // TODO: Replace with catch {} when available.
    let r = (|| -> Result<()> {
        let f = opt.env.open_writable_file(Path::new(&filename))?;
        let f = BufWriter::new(f);
        // 使用 TableBuilder 将键值对按 SSTable 格式（包含索引、块结构等）写入磁盘文件
        let mut builder = TableBuilder::new(opt.clone(), f);
        // 通过迭代器（from）遍历内存中的键值对。
        while from.advance() {
            if let Some((key_bytes, val_bytes)) = from.current() {
                kbuf = key_bytes.to_vec();
                vbuf = val_bytes.to_vec();
                if firstkey.is_none() {
                    firstkey = Some(kbuf.clone());
                }
                builder.add(&kbuf, &vbuf)?;
            } else {
                panic!("Iterator should be valid here");
            }
        }
        builder.finish()?;
        Ok(())
    })();
    // 如果写入过程失败，删除已创建的文件，避免残留无效文件。
    if let Err(e) = r {
        let _ = opt.env.delete(Path::new(&filename));
        return Err(e);
    }

    // 成功写入后，创建 FileMetaData 结构体，记录：
    // 文件编号（num）、大小（size）。
    // 最小键（smallest）和最大键（largest，来自最后一个写入的键）。
    // 若没有数据（空表），则删除文件，返回空元数据。
    let mut md = FileMetaData::default();
    match firstkey {
        None => {
            let _ = opt.env.delete(Path::new(&filename));
        }
        Some(key) => {
            md.num = num; //文件编号
            md.size = opt.env.size_of(Path::new(&filename))?; //大小
            md.smallest = key.into(); //最小键
            md.largest = kbuf.into(); //最大键
        }
    }
    Ok(md)
}

/// 把日志文件名后面追加 .log 后缀
fn log_file_name(db: &Path, num: FileNum) -> PathBuf {
    db.join(format!("{:06}.log", num))
}

fn lock_file_name(db: &Path) -> PathBuf {
    db.join("LOCK")
}

/// open_info_log opens an info log file in the given database. It transparently returns a
/// /dev/null logger in case the open fails.
fn open_info_log<E: Env + ?Sized, P: AsRef<Path>>(env: &E, db: P) -> Logger {
    let db = db.as_ref();
    let logfilename = db.join("LOG");
    let oldlogfilename = db.join("LOG.old");
    let _ = env.mkdir(Path::new(db));
    if let Ok(e) = env.exists(Path::new(&logfilename)) {
        if e {
            let _ = env.rename(Path::new(&logfilename), Path::new(&oldlogfilename));
        }
    }
    if let Ok(w) = env.open_writable_file(Path::new(&logfilename)) {
        Logger(w)
    } else {
        Logger(Box::new(io::sink()))
    }
}

#[cfg(test)]
pub mod testutil {
    use super::*;

    use crate::version::testutil::make_version;

    /// build_db creates a database filled with the tables created by make_version().
    pub fn build_db() -> (DB, Options) {
        let name = "db";
        let (v, mut opt) = make_version();
        opt.reuse_logs = false;
        opt.reuse_manifest = false;
        let mut ve = VersionEdit::new();
        ve.set_comparator_name(opt.cmp.id());
        ve.set_log_num(0);
        // 9 files + 1 manifest we write below.
        ve.set_next_file(11);
        // 30 entries in these tables.
        ve.set_last_seq(30);

        for l in 0..NUM_LEVELS {
            for f in &v.files[l] {
                ve.add_file(l, f.borrow().clone());
            }
        }

        let manifest = manifest_file_name(name, 10);
        let manifest_file = opt.env.open_writable_file(Path::new(&manifest)).unwrap();
        let mut lw = LogWriter::new(manifest_file);
        lw.add_record(&ve.encode()).unwrap();
        lw.flush().unwrap();
        set_current_file(opt.env.as_ref().as_ref(), name, 10).unwrap();

        (DB::open(name, opt.clone()).unwrap(), opt)
    }

    /// set_file_to_compact ensures that the specified table file will be compacted next.
    pub fn set_file_to_compact(db: &mut DB, num: FileNum) {
        let v = db.current();
        let mut v = v.borrow_mut();

        let mut ftc = None;
        for l in 0..NUM_LEVELS {
            for f in &v.files[l] {
                if f.borrow().num == num {
                    ftc = Some((f.clone(), l));
                }
            }
        }
        if let Some((f, l)) = ftc {
            v.file_to_compact = Some(f);
            v.file_to_compact_lvl = l;
        } else {
            panic!("file number not found");
        }
    }
}

#[cfg(test)]
mod tests {
    use super::testutil::{build_db, set_file_to_compact};
    use super::*;

    use crate::error::Status;
    use crate::key_types::LookupKey;
    use crate::mem_env::MemEnv;
    use crate::options;
    use crate::test_util::LdbIteratorIter;
    use crate::version::testutil::make_version;

    #[test]
    fn test_db_impl_open_info_log() {
        let e = MemEnv::new();
        {
            let l = Some(share(open_info_log(&e, "abc")));
            assert!(e.exists(&Path::new("abc").join("LOG")).unwrap());
            log!(l, "hello {}", "world");
            assert_eq!(12, e.size_of(&Path::new("abc").join("LOG")).unwrap());
        }
        {
            let l = Some(share(open_info_log(&e, "abc")));
            assert!(e.exists(&Path::new("abc").join("LOG.old")).unwrap());
            assert!(e.exists(&Path::new("abc").join("LOG")).unwrap());
            assert_eq!(12, e.size_of(&Path::new("abc").join("LOG.old")).unwrap());
            assert_eq!(0, e.size_of(&Path::new("abc").join("LOG")).unwrap());
            log!(l, "something else");
            log!(l, "and another {}", 1);

            let mut s = String::new();
            let mut r = e
                .open_sequential_file(&Path::new("abc").join("LOG"))
                .unwrap();
            r.read_to_string(&mut s).unwrap();
            assert_eq!("something else\nand another 1\n", &s);
        }
    }

    fn build_memtable() -> MemTable {
        let mut mt = MemTable::new(options::for_test().cmp);
        let mut i = 1;
        for k in ["abc", "def", "ghi", "jkl", "mno", "aabc", "test123"].iter() {
            mt.add(i, ValueType::TypeValue, k.as_bytes(), b"looooongval");
            i += 1;
        }
        mt
    }

    #[test]
    fn test_db_impl_init() {
        // A sanity check for recovery and basic persistence.
        let opt = options::for_test();
        let env = opt.env.clone();

        // Several test cases with different options follow. The printlns can eventually be
        // removed.

        {
            let mut opt = opt.clone();
            opt.reuse_manifest = false;
            let _ = DB::open("otherdb", opt.clone()).unwrap();

            eprintln!(
                "children after: {:?}",
                env.children(Path::new("otherdb")).unwrap()
            );
            assert!(env.exists(&Path::new("otherdb").join("CURRENT")).unwrap());
            // Database is initialized and initial manifest reused.
            assert!(!env
                .exists(&Path::new("otherdb").join("MANIFEST-000001"))
                .unwrap());
            assert!(env
                .exists(&Path::new("otherdb").join("MANIFEST-000002"))
                .unwrap());
            assert!(env
                .exists(&Path::new("otherdb").join("000003.log"))
                .unwrap());
        }

        {
            let mut opt = opt.clone();
            opt.reuse_manifest = true;
            let mut db = DB::open("db", opt.clone()).unwrap();

            eprintln!(
                "children after: {:?}",
                env.children(&Path::new("db").join("")).unwrap()
            );
            assert!(env.exists(&Path::new("db").join("CURRENT")).unwrap());
            // Database is initialized and initial manifest reused.
            assert!(env
                .exists(&Path::new("db").join("MANIFEST-000001"))
                .unwrap());
            assert!(env.exists(&Path::new("db").join("LOCK")).unwrap());
            assert!(env.exists(&Path::new("db").join("000003.log")).unwrap());

            db.put(b"abc", b"def").unwrap();
            db.put(b"abd", b"def").unwrap();
        }

        {
            eprintln!(
                "children before: {:?}",
                env.children(&Path::new("db").join("")).unwrap()
            );
            let mut opt = opt.clone();
            opt.reuse_manifest = false;
            opt.reuse_logs = false;
            let mut db = DB::open("db", opt.clone()).unwrap();

            eprintln!(
                "children after: {:?}",
                env.children(&Path::new("db").join("")).unwrap()
            );
            // Obsolete manifest is deleted.
            assert!(!env
                .exists(&Path::new("db").join("MANIFEST-000001"))
                .unwrap());
            // New manifest is created.
            assert!(env
                .exists(&Path::new("db").join("MANIFEST-000002"))
                .unwrap());
            // Obsolete log file is deleted.
            assert!(!env.exists(&Path::new("db").join("000003.log")).unwrap());
            // New L0 table has been added.
            assert!(env.exists(&Path::new("db").join("000003.ldb")).unwrap());
            assert!(env.exists(&Path::new("db").join("000004.log")).unwrap());
            // Check that entry exists and is correct. Phew, long call chain!
            let current = db.current();
            log!(opt.log, "files: {:?}", current.borrow().files);
            assert_eq!(
                b"def",
                current
                    .borrow_mut()
                    .get(LookupKey::new(b"abc", 1).internal_key())
                    .unwrap()
                    .unwrap()
                    .0
                    .as_ref()
            );
            db.put(b"abe", b"def").unwrap();
        }

        {
            eprintln!(
                "children before: {:?}",
                env.children(Path::new("db")).unwrap()
            );
            // reuse_manifest above causes the old manifest to be deleted as obsolete, but no new
            // manifest is written. CURRENT becomes stale.
            let mut opt = opt.clone();
            opt.reuse_logs = true;
            let db = DB::open("db", opt).unwrap();

            eprintln!(
                "children after: {:?}",
                env.children(Path::new("db")).unwrap()
            );
            assert!(!env
                .exists(&Path::new("db").join("MANIFEST-000001"))
                .unwrap());
            assert!(env
                .exists(&Path::new("db").join("MANIFEST-000002"))
                .unwrap());
            assert!(!env
                .exists(&Path::new("db").join("MANIFEST-000005"))
                .unwrap());
            assert!(env.exists(&Path::new("db").join("000004.log")).unwrap());
            // 000004 should be reused, no new log file should be created.
            assert!(!env.exists(&Path::new("db").join("000006.log")).unwrap());
            // Log is reused, so memtable should contain last written entry from above.
            assert_eq!(1, db.mem.len());
            assert_eq!(b"def", &*db.mem.get(&LookupKey::new(b"abe", 3)).0.unwrap());
        }
    }

    #[test]
    fn test_db_impl_compact_range() {
        let (mut db, opt) = build_db();
        let env = &opt.env;

        eprintln!(
            "children before: {:?}",
            env.children(&Path::new("db").join("")).unwrap()
        );
        db.compact_range(b"aaa", b"dba").unwrap();
        eprintln!(
            "children after: {:?}",
            env.children(&Path::new("db").join("")).unwrap()
        );

        assert_eq!(
            250,
            opt.env
                .size_of(&Path::new("db").join("000007.ldb"))
                .unwrap()
        );
        assert_eq!(
            200,
            opt.env
                .size_of(&Path::new("db").join("000008.ldb"))
                .unwrap()
        );
        assert_eq!(
            200,
            opt.env
                .size_of(&Path::new("db").join("000009.ldb"))
                .unwrap()
        );
        assert_eq!(
            435,
            opt.env
                .size_of(&Path::new("db").join("000015.ldb"))
                .unwrap()
        );

        assert!(!opt.env.exists(&Path::new("db").join("000001.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000002.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000004.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000005.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000006.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000013.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000014.ldb")).unwrap());

        assert_eq!(b"val1".to_vec(), db.get(b"aaa").unwrap());
        assert_eq!(b"val2".to_vec(), db.get(b"cab").unwrap());
        assert_eq!(b"val3".to_vec(), db.get(b"aba").unwrap());
        assert_eq!(b"val3".to_vec(), db.get(b"fab").unwrap());
    }

    #[test]
    fn test_db_impl_compact_range_memtable() {
        let (mut db, opt) = build_db();
        let env = &opt.env;

        db.put(b"xxx", b"123").unwrap();

        eprintln!(
            "children before: {:?}",
            env.children(Path::new("db")).unwrap()
        );
        db.compact_range(b"aaa", b"dba").unwrap();
        eprintln!(
            "children after: {:?}",
            env.children(Path::new("db")).unwrap()
        );

        assert_eq!(
            250,
            opt.env
                .size_of(&Path::new("db").join("000007.ldb"))
                .unwrap()
        );
        assert_eq!(
            200,
            opt.env
                .size_of(&Path::new("db").join("000008.ldb"))
                .unwrap()
        );
        assert_eq!(
            200,
            opt.env
                .size_of(&Path::new("db").join("000009.ldb"))
                .unwrap()
        );
        assert_eq!(
            182,
            opt.env
                .size_of(&Path::new("db").join("000014.ldb"))
                .unwrap()
        );
        assert_eq!(
            435,
            opt.env
                .size_of(&Path::new("db").join("000017.ldb"))
                .unwrap()
        );

        assert!(!opt.env.exists(&Path::new("db").join("000001.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000002.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000003.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000004.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000005.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000006.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000015.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000016.ldb")).unwrap());

        assert_eq!(b"val1".to_vec(), db.get(b"aaa").unwrap());
        assert_eq!(b"val2".to_vec(), db.get(b"cab").unwrap());
        assert_eq!(b"val3".to_vec(), db.get(b"aba").unwrap());
        assert_eq!(b"val3".to_vec(), db.get(b"fab").unwrap());
        assert_eq!(b"123".to_vec(), db.get(b"xxx").unwrap());
    }

    #[allow(unused_variables)]
    #[test]
    fn test_db_impl_locking() {
        let opt = options::for_test();
        let db = DB::open("db", opt.clone()).unwrap();
        let want_err = Status::new(
            StatusCode::LockError,
            "database lock is held by another instance",
        );
        assert_eq!(want_err, DB::open("db", opt.clone()).err().unwrap());
    }

    #[test]
    fn test_db_impl_build_table() {
        let mut opt = options::for_test();
        opt.block_size = 128;
        let mt = build_memtable();

        let f = build_table("db", &opt, mt.iter(), 123).unwrap();
        let path = &Path::new("db").join("000123.ldb");

        assert_eq!(LookupKey::new(b"aabc", 6).internal_key(), &f.smallest);
        assert_eq!(LookupKey::new(b"test123", 7).internal_key(), &f.largest);
        assert_eq!(379, f.size);
        assert_eq!(123, f.num);
        assert!(opt.env.exists(path).unwrap());

        {
            // Read table back in.
            let mut tc = TableCache::new("db", opt.clone(), share(Cache::new(128)), 100);
            let tbl = tc.get_table(123).unwrap();
            assert_eq!(mt.len(), LdbIteratorIter::wrap(&mut tbl.iter()).count());
        }

        {
            // Corrupt table; make sure it doesn't load fully.
            let mut buf = vec![];
            opt.env
                .open_sequential_file(path)
                .unwrap()
                .read_to_end(&mut buf)
                .unwrap();
            buf[150] += 1;
            opt.env
                .open_writable_file(path)
                .unwrap()
                .write_all(&buf)
                .unwrap();

            let mut tc = TableCache::new("db", opt.clone(), share(Cache::new(128)), 100);
            let tbl = tc.get_table(123).unwrap();
            // The last two entries are skipped due to the corruption above.
            assert_eq!(
                5,
                LdbIteratorIter::wrap(&mut tbl.iter())
                    .map(|v| eprintln!("{:?}", v))
                    .count()
            );
        }
    }

    #[allow(unused_variables)]
    #[test]
    fn test_db_impl_build_db_sanity() {
        let db = build_db().0;
        let env = &db.opt.env;
        let name = &db.name;

        assert!(env.exists(Path::new(&log_file_name(name, 12))).unwrap());
    }

    #[test]
    fn test_db_impl_get_from_table_with_snapshot() {
        let mut db = build_db().0;

        assert_eq!(30, db.vset.borrow().last_seq);

        // seq = 31
        db.put(b"xyy", b"123").unwrap();
        let old_ss = db.get_snapshot();
        // seq = 32
        db.put(b"xyz", b"123").unwrap();
        db.flush().unwrap();
        assert!(db.get_at(&old_ss, b"xyy").unwrap().is_some());
        assert!(db.get_at(&old_ss, b"xyz").unwrap().is_none());

        // memtable get
        assert_eq!(b"123", db.get(b"xyz").unwrap().as_ref());
        assert!(db.get_internal(31, b"xyy").unwrap().is_some());
        assert!(db.get_internal(32, b"xyy").unwrap().is_some());

        assert!(db.get_internal(31, b"xyz").unwrap().is_none());
        assert!(db.get_internal(32, b"xyz").unwrap().is_some());

        // table get
        assert_eq!(b"val2", db.get(b"eab").unwrap().as_ref());
        assert!(db.get_internal(3, b"eab").unwrap().is_none());
        assert!(db.get_internal(32, b"eab").unwrap().is_some());

        {
            let ss = db.get_snapshot();
            assert_eq!(b"val2", db.get_at(&ss, b"eab").unwrap().unwrap().as_ref());
        }

        // from table.
        assert_eq!(b"val2", db.get(b"cab").unwrap().as_ref());
    }

    #[test]
    fn test_db_impl_delete() {
        let mut db = build_db().0;

        db.put(b"xyy", b"123").unwrap();
        db.put(b"xyz", b"123").unwrap();

        assert!(db.get(b"xyy").is_some());
        assert!(db.get(b"gaa").is_some());

        // Delete one memtable entry and one table entry.
        db.delete(b"xyy").unwrap();
        db.delete(b"gaa").unwrap();

        assert!(db.get(b"xyy").is_none());
        assert!(db.get(b"gaa").is_none());
        assert!(db.get(b"xyz").is_some());
    }

    #[test]
    fn test_db_impl_compact_single_file() {
        let mut db = build_db().0;
        set_file_to_compact(&mut db, 4);
        db.maybe_do_compaction().unwrap();

        let env = &db.opt.env;
        let name = &db.name;
        assert!(!env.exists(Path::new(&table_file_name(name, 3))).unwrap());
        assert!(!env.exists(Path::new(&table_file_name(name, 4))).unwrap());
        assert!(!env.exists(Path::new(&table_file_name(name, 5))).unwrap());
        assert!(env.exists(Path::new(&table_file_name(name, 13))).unwrap());
    }

    #[test]
    fn test_db_impl_compaction_trivial_move() {
        let mut db = DB::open("db", options::for_test()).unwrap();

        db.put(b"abc", b"xyz").unwrap();
        db.put(b"ab3", b"xyz").unwrap();
        db.put(b"ab0", b"xyz").unwrap();
        db.put(b"abz", b"xyz").unwrap();
        assert_eq!(4, db.mem.len());
        let mut imm = MemTable::new(db.opt.cmp.clone());
        mem::swap(&mut imm, &mut db.mem);
        db.imm = Some(imm);
        db.compact_memtable().unwrap();

        eprintln!(
            "children after: {:?}",
            db.opt.env.children(Path::new("db")).unwrap()
        );
        assert!(db
            .opt
            .env
            .exists(&Path::new("db").join("000004.ldb"))
            .unwrap());

        {
            let v = db.current();
            let mut v = v.borrow_mut();
            v.file_to_compact = Some(v.files[2][0].clone());
            v.file_to_compact_lvl = 2;
        }

        db.maybe_do_compaction().unwrap();

        {
            let v = db.current();
            let v = v.borrow_mut();
            assert_eq!(1, v.files[3].len());
        }
    }

    #[test]
    fn test_db_impl_memtable_compaction() {
        let mut opt = options::for_test();
        opt.write_buffer_size = 25;
        let mut db = DB::new("db", opt);

        // Fill up memtable.
        db.mem = build_memtable();

        // Trigger memtable compaction.
        db.make_room_for_write(true).unwrap();
        assert_eq!(0, db.mem.len());
        assert!(db
            .opt
            .env
            .exists(&Path::new("db").join("000002.log"))
            .unwrap());
        assert!(db
            .opt
            .env
            .exists(&Path::new("db").join("000003.ldb"))
            .unwrap());
        assert_eq!(
            351,
            db.opt
                .env
                .size_of(&Path::new("db").join("000003.ldb"))
                .unwrap()
        );
        assert_eq!(
            7,
            LdbIteratorIter::wrap(&mut db.cache.borrow_mut().get_table(3).unwrap().iter()).count()
        );
    }

    #[test]
    fn test_db_impl_compaction() {
        let mut db = build_db().0;
        let v = db.current();
        v.borrow_mut().compaction_score = Some(2.0);
        v.borrow_mut().compaction_level = Some(1);

        db.maybe_do_compaction().unwrap();

        assert!(!db
            .opt
            .env
            .exists(&Path::new("db").join("000003.ldb"))
            .unwrap());
        assert!(db
            .opt
            .env
            .exists(&Path::new("db").join("000013.ldb"))
            .unwrap());
        assert_eq!(
            345,
            db.opt
                .env
                .size_of(&Path::new("db").join("000013.ldb"))
                .unwrap()
        );

        // New current version.
        let v = db.current();
        assert_eq!(0, v.borrow().files[1].len());
        assert_eq!(2, v.borrow().files[2].len());
    }

    #[test]
    fn test_db_impl_compaction_trivial() {
        let (mut v, opt) = make_version();

        let to_compact = v.files[2][0].clone();
        v.file_to_compact = Some(to_compact);
        v.file_to_compact_lvl = 2;

        let mut db = DB::new("db", opt.clone());
        db.vset.borrow_mut().add_version(v);
        db.vset.borrow_mut().next_file_num = 10;

        db.maybe_do_compaction().unwrap();

        assert!(opt.env.exists(&Path::new("db").join("000006.ldb")).unwrap());
        assert!(!opt.env.exists(&Path::new("db").join("000010.ldb")).unwrap());
        assert_eq!(
            218,
            opt.env
                .size_of(&Path::new("db").join("000006.ldb"))
                .unwrap()
        );

        let v = db.current();
        assert_eq!(1, v.borrow().files[2].len());
        assert_eq!(3, v.borrow().files[3].len());
    }

    #[test]
    fn test_db_impl_compaction_state_cleanup() {
        let env: Box<dyn Env> = Box::new(MemEnv::new());
        let name = "db";

        let stuff = b"abcdefghijkl";
        env.open_writable_file(&Path::new("db").join("000001.ldb"))
            .unwrap()
            .write_all(stuff)
            .unwrap();
        let fmd = FileMetaData {
            num: 1,
            ..Default::default()
        };

        let mut cs = CompactionState::new(Compaction::new(&options::for_test(), 2, None), 12);
        cs.outputs = vec![fmd];
        cs.cleanup(env.as_ref(), name);

        assert!(!env.exists(&Path::new("db").join("000001.ldb")).unwrap());
    }

    #[test]
    fn test_db_impl_open_close_reopen() {
        let opt;
        {
            let mut db = build_db().0;
            opt = db.opt.clone();
            db.put(b"xx1", b"111").unwrap();
            db.put(b"xx2", b"112").unwrap();
            db.put(b"xx3", b"113").unwrap();
            db.put(b"xx4", b"114").unwrap();
            db.put(b"xx5", b"115").unwrap();
            db.delete(b"xx2").unwrap();
        }

        {
            let mut db = DB::open("db", opt.clone()).unwrap();
            db.delete(b"xx5").unwrap();
        }

        {
            let mut db = DB::open("db", opt.clone()).unwrap();

            assert_eq!(None, db.get(b"xx5"));

            let ss = db.get_snapshot();
            db.put(b"xx4", b"222").unwrap();
            let ss2 = db.get_snapshot();

            assert_eq!(
                Some(b"113".to_vec()),
                db.get_at(&ss, b"xx3").unwrap().map(|b| b.to_vec())
            );
            assert_eq!(None, db.get_at(&ss, b"xx2").unwrap());
            assert_eq!(None, db.get_at(&ss, b"xx5").unwrap());

            assert_eq!(
                Some(b"114".to_vec()),
                db.get_at(&ss, b"xx4").unwrap().map(|b| b.to_vec())
            );
            assert_eq!(
                Some(b"222".to_vec()),
                db.get_at(&ss2, b"xx4").unwrap().map(|b| b.to_vec())
            );
        }

        {
            let mut db = DB::open("db", opt).unwrap();

            let ss = db.get_snapshot();
            assert_eq!(
                Some(b"113".to_vec()),
                db.get_at(&ss, b"xx3").unwrap().map(|b| b.to_vec())
            );
            assert_eq!(
                Some(b"222".to_vec()),
                db.get_at(&ss, b"xx4").unwrap().map(|b| b.to_vec())
            );
            assert_eq!(None, db.get_at(&ss, b"xx2").unwrap());
        }
    }

    #[test]
    fn test_drop_memtable() {
        let mut db = DB::open("db", options::for_test()).unwrap();
        let mut cnt = 0;
        let mut buf: Vec<u8> = Vec::with_capacity(8);
        let entries_per_batch = 1;
        let max_num = 100000;
        while cnt < max_num {
            let mut write_batch = WriteBatch::new();
            for i in 0..entries_per_batch {
                buf.clear();
                buf.extend_from_slice(format!("{}-{}", cnt, i).as_bytes());
                write_batch.put(buf.as_slice(), buf.as_slice());
            }
            db.write(write_batch, false).unwrap();
            cnt += entries_per_batch;
        }
    }
}
